package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TrafficPageViewBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Iterator;

//按照版本代码、渠道、地区和是否为新用户进行分组，并在每小时的滚动时间窗口内统计页面浏览量
// 页面浏览时长、页面浏览数、会话数和独立访客
public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficVcChArIsNewPageViewWindow().start(Constant.TOPIC_DWD_TRAFFIC_PAGE,Constant.DWS_TRAFFIC_VC_CH_AR_IS_NEW_PAGE_VIEW_WINDOW,4,10022);
    }
    // 12.2流量域版本-渠道-地区-访客类别粒度页面浏览各窗口汇总表
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        // 1、ETL数据
        SingleOutputStreamOperator<JSONObject> etlStream = etl(dataStreamSource);
        // 2、Keyby
        SingleOutputStreamOperator<TrafficPageViewBean> processStream  = extracted(etlStream);
        // 3、添加水位线
        KeyedStream<TrafficPageViewBean, String> keyedStream = getKeyedStream(processStream);
        // 4、开窗聚合
        SingleOutputStreamOperator<TrafficPageViewBean> reduceStream = getReduce(keyedStream);
        // 5、写入到Doris
//        reduceStream.sinkTo(FlinkSinkUtil.getDorisSink())\


        // 写入Doris一定要开启CK
        reduceStream.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DWS_TRAFFIC_VC_CH_AR_IS_NEW_PAGE_VIEW_WINDOW));

    }

    private SingleOutputStreamOperator<TrafficPageViewBean> getReduce(KeyedStream<TrafficPageViewBean, String> keyedStream) {
        return keyedStream.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {

                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        return value1;

                    }
                }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                        // 最终只有一条数据
                        Iterator<TrafficPageViewBean> iterator = iterable.iterator();
                        String start = DateFormatUtil.tsToDateTime(timeWindow.getStart());
                        String end = DateFormatUtil.tsToDateTime(timeWindow.getEnd());//开窗的结束时间
                        while (iterator.hasNext()) {
                            TrafficPageViewBean trafficPageViewBean = iterator.next();
                            trafficPageViewBean.setStt(start);
                            trafficPageViewBean.setEdt(end);
                            trafficPageViewBean.setCur_date(DateFormatUtil.tsToDate(System.currentTimeMillis()));
                            collector.collect(trafficPageViewBean);
                        }
                    }
                });
    }

    private KeyedStream<TrafficPageViewBean, String> getKeyedStream(SingleOutputStreamOperator<TrafficPageViewBean> processStream) {
        return processStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
            @Override
            public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                return trafficPageViewBean.getTs();
            }
        })).keyBy(new KeySelector<TrafficPageViewBean, String>() {
            @Override
            public String getKey(TrafficPageViewBean trafficPageViewBean) throws Exception {
                return trafficPageViewBean.getVc() + "-" + trafficPageViewBean.getCh() + "-" + trafficPageViewBean.getAr() + "-" + trafficPageViewBean.getIsNew();
            }
        });
    }

    private SingleOutputStreamOperator<TrafficPageViewBean> extracted(SingleOutputStreamOperator<JSONObject> etlStream) {
        return  etlStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("mid");
            }
        }).process(new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
            private ValueState<String> lastDtState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("last_dt", String.class);
                valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.hours(24))// 默认更新方式---OnCreateAndWrite
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                ;
                lastDtState = getRuntimeContext().getState(valueStateDescriptor);
                // TTL
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>.Context context, Collector<TrafficPageViewBean> collector) throws Exception {

                Long ts = jsonObject.getLong("ts");
                String curDate = DateFormatUtil.tsToDate(ts);
                String value = lastDtState.value();
                Long uvCt = 0L;
                Long svCt = 0L;

                if (!curDate.equals(value)) {
                    uvCt = 1L;
                    lastDtState.update(curDate);
                }



                JSONObject page = jsonObject.getJSONObject("page");
                String last_page_id = page.getString("last_page_id");
                if (last_page_id == null) {
                    // 会话数
                    svCt = 1L;
                }

                JSONObject common = jsonObject.getJSONObject("common");
                collector.collect(TrafficPageViewBean.builder()
                        .ar(common.getString("ar"))
                        .ch(common.getString("ch"))
                        .vc(common.getString("vc"))
                        .isNew(common.getString("is_new"))
                        .uvCt(uvCt)
                        .svCt(svCt)
                        .pvCt(1L)
                        .durSum(page.getLong("during_time"))
                        .ts(ts)
                        .sid(common.getString("sid"))
                        .build());
            }
        });
    }


    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> kafkaSource) {
        return kafkaSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");
                String ts = jsonObject.getString("ts");
                if (common != null && page != null) {
                    String mid = common.getString("mid");// 设备MID
                    if (mid != null && ts != null) {
                        collector.collect(jsonObject);
                    }
                }
            }
        });
    }
}
